package com.ruoyi.job.thread;

import com.alibaba.fastjson.JSONObject;
import com.api.equipment.RemoteEquipmentService;
import com.api.equipment.domain.LgEquipmentError;
import com.api.equipment.domain.LgMachine;
import com.api.equipment.domain.LgTemperature;
import com.ruoyi.common.core.utils.SpringUtils;
import com.ruoyi.job.redis.RedisUtil;
import com.ruoyi.job.util.MailUtil;
import org.eclipse.paho.client.mqttv3.*;

import java.math.BigDecimal;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author: lek
 * @Description: TODO
 * @Date: 2025/3/16 16:41
 * @Version: 1.0
 */

public class TemThread extends Thread {
    private String serial;
    private String cmd;
    private Random random = new Random();
    String brokerUrl = "tcp://124.223.12.211:1883";
    String pubclientId = "temperature_publisher";
    String subclientId = "temperature_subscribe";
    String topic = "sensor/temperature/";
    int pubQos = 1;
    int subQos = 1;
    public static ConcurrentHashMap<String, Thread> consoleThreadMap = new ConcurrentHashMap<String, Thread>();
    public ConcurrentHashMap<String, JSONObject> subMessage = new ConcurrentHashMap<String, JSONObject>();

    private volatile boolean running = true; // 新增运行状态标志
    private MqttClient publisherClient;
    private MqttClient subscriberClient;
    // 用于记录上次发送邮件的时间
    private long lastMailSendTime = 0;

    private MailUtil mailUtil = SpringUtils.getBean(MailUtil.class);

    private RemoteEquipmentService remoteEquipmentService = SpringUtils.getBean(RemoteEquipmentService.class);

    RedisUtil redisUtil = SpringUtils.getBean(RedisUtil.class);

    public String getSerial() {
        return serial;
    }

    public void setSerial(String serial) {
        this.serial = serial;
    }

    public static ConcurrentHashMap<String, Thread> getConsoleThreadMap() {
        return consoleThreadMap;
    }

    public static void setConsoleThreadMap(ConcurrentHashMap<String, Thread> consoleThreadMap) {
        TemThread.consoleThreadMap = consoleThreadMap;
    }

    public String getCmd() {
        return cmd;
    }

    public void setCmd(String cmd) {
        this.cmd = cmd;
    }

    public ConcurrentHashMap<String, JSONObject> getSubMessage() {
        return subMessage;
    }

    public void setSubMessage(ConcurrentHashMap<String, JSONObject> subMessage) {
        this.subMessage = subMessage;
    }

    @Override
    public void run() {
        try {
            if ("publish".equals(cmd)) {
                consoleThreadMap.put("publish" + serial, Thread.currentThread());
                publish();
            } else if ("subscribe".equals(cmd)) {
                consoleThreadMap.put("subscribe" + serial, Thread.currentThread());
                subscribe();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 根据命令类型移除对应的线程引用
            if ("publish".equals(cmd)) {
                consoleThreadMap.remove("publish" + serial);
            } else if ("subscribe".equals(cmd)) {
                consoleThreadMap.remove("subscribe" + serial);
            }
            if(redisUtil != null){
                //jedisPool.returnResourceObject(jedis);  // 已废弃，推荐使用jedis.close()方法
                redisUtil.close(redisUtil.getJedis());// 关闭redis连接,还回redis连接池中；
            }
        }
    }

    // 发布方法改进
    public void publish() throws MqttException {
        try {
            publisherClient = new MqttClient(brokerUrl, pubclientId + "_" + serial+ "_" + System.currentTimeMillis());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(false);
            publisherClient.connect(options);
            // 在连接后检查连接状态
            if (publisherClient.isConnected()) {
                System.out.println("Publisher connected successfully");
            } else {
                System.out.println("Publisher connection failed");
            }
            String targetTopic = topic + serial;

            // 设定洛阳为起始位置
            double latitude = 34.6186;
            double longitude = 112.4535;


            while (running && !Thread.currentThread().isInterrupted()) {
                // 构造并发送消息
                JSONObject sensorData = new JSONObject();
                double temperature = random.nextInt(5) + random.nextDouble();
                sensorData.put("temperature", temperature);
//                sensorData.put("temperature",1.0 + random.nextInt(2));
                sensorData.put("humidity", 50 + random.nextInt(20));
                // 生成相邻合理的 GPS 坐标，确保移动范围合理（0.0005 ~ 0.0015 度）
                // 增加经纬度变动范围：-0.005 到 +0.005
                latitude += (random.nextDouble() * 0.01 - 0.005); // -0.005 ~ +0.005
                longitude += (random.nextDouble() * 0.01 - 0.005); // -0.005 ~ +0.005
                // 保证 GPS 坐标仍在洛阳附近，但可以适当放宽限制（例如扩大范围）
                latitude = Math.max(34.4000, Math.min(34.8000, latitude));  // 限制纬度范围：扩大到 34.4000 ~ 34.8000
                longitude = Math.max(112.2000, Math.min(112.8000, longitude)); // 限制经度范围：扩大到 112.2000 ~ 112.8000

                // 使用 BigDecimal 保留六位小数
                sensorData.put("latitude", new BigDecimal(latitude).setScale(6, BigDecimal.ROUND_HALF_UP).doubleValue());
                sensorData.put("longitude", new BigDecimal(longitude).setScale(6, BigDecimal.ROUND_HALF_UP).doubleValue());

                MqttMessage message = new MqttMessage(sensorData.toJSONString().getBytes());
                message.setQos(pubQos);
                publisherClient.publish(targetTopic, message);
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    running = false;
                    break;
                }
            }
        } finally {
            if (publisherClient != null && publisherClient.isConnected()) {
                try {
                    publisherClient.disconnect();
                    publisherClient.close();
                } catch (MqttException e) {
                    System.out.println(e.getMessage());
                }
            }
//            // 仅关闭发布客户端
//            if (publisherClient != null && publisherClient.isConnected()) {
//                try {
//                    publisherClient.disconnectForcibly(1000);
//                    publisherClient.close();
//                } catch (MqttException e) {
//                    e.printStackTrace();
//                }
//            }
        }
    }

    // 订阅方法改进
    public void subscribe() throws MqttException {
        try {
            subscriberClient = new MqttClient(brokerUrl, subclientId + "_" + serial + "_" + System.currentTimeMillis());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(false);
            subscriberClient.connect(options);
            subscriberClient.subscribe(topic + serial, subQos);
            subscriberClient.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
//                System.out.println("Connection lost: " + cause.getMessage());
                    running = false; // 触发终止
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) {
                    // 处理消息逻辑...
                    String payload = new String(message.getPayload());
                    JSONObject data = JSONObject.parseObject(payload); // Fastjson解析
                    System.out.println("设备标号"+serial);
                    LgTemperature lgTemperature = remoteEquipmentService.getTemInfo(Integer.valueOf(serial));
                    lgTemperature.setCurrentTemperature(data.getInteger("temperature"));
                    lgTemperature.setCurrentHumidity(data.getInteger("humidity"));
                    lgTemperature.setLatitudeX(data.getDouble("latitude"));
                    lgTemperature.setLatitudeY(data.getDouble("longitude"));
                    remoteEquipmentService.setTem(lgTemperature);

                    //判断当前温湿度是否合规  不合规存入异常表
                    if (data.getInteger("temperature") > lgTemperature.getLgColdType().getMaxTemp()  || data.getInteger("temperature") < lgTemperature.getLgColdType().getMinTemp()) {
                        LgEquipmentError lgEquipmentError = new LgEquipmentError();
                        lgEquipmentError.setDeviceNumber(lgTemperature.getDeviceNumber());
                        lgEquipmentError.setAlarmTime(new Timestamp(new Date().getTime()).toString());
                        lgEquipmentError.setAlarmMessage("温度异常");
                        lgEquipmentError.setAlarmCount(1);
                        lgEquipmentError.setEquipmentType("0");
                        lgEquipmentError.setIsProcessed(0);
                        remoteEquipmentService.insertEquipError(lgEquipmentError);

                        System.out.println(lgTemperature.getLgDriver().getEmail());
                        // 获取当前时间
                        long currentTime = new Date().getTime();

                        // 判断当前时间与上次发送邮件时间的间隔是否大于 4 分钟 (240000 毫秒)
                        if (currentTime - lastMailSendTime > 240000) {
                            // 发送邮件
                            mailUtil.sendSimpleMail(lgTemperature.getLgDriver().getEmail(), "温度异常", "温度异常");
                            // 更新上次发送邮件的时间
                            lastMailSendTime = currentTime;
                        }
                        //给管理员发送邮件
//                        mailUtil.sendSimpleMail(lgTemperature.getLgDriver().getEmail(), "温度异常", "温度异常");
                    }
                    if (data.getInteger("humidity") > lgTemperature.getLgColdType().getMaxHumidity() || data.getInteger("humidity") < lgTemperature.getLgColdType().getMinHumidity()){
                        LgEquipmentError lgEquipmentError = new LgEquipmentError();
                        lgEquipmentError.setDeviceNumber(lgTemperature.getDeviceNumber());
                        lgEquipmentError.setAlarmTime(new Timestamp(new Date().getTime()).toString());
                        lgEquipmentError.setAlarmMessage("湿度异常");
                        lgEquipmentError.setAlarmCount(1);
                        lgEquipmentError.setEquipmentType("2");
                        lgEquipmentError.setIsProcessed(0);
                        remoteEquipmentService.insertEquipError(lgEquipmentError);
                        System.out.println(lgTemperature.getLgDriver().getEmail());
                        // 获取当前时间
                        long currentTime = new Date().getTime();

                        // 判断当前时间与上次发送邮件时间的间隔是否大于 4 分钟 (240000 毫秒)
                        if (currentTime - lastMailSendTime > 240000) {
                            //给管理员发送邮件
                            mailUtil.sendSimpleMail(lgTemperature.getLgDriver().getEmail(), "温度异常", "温度异常");
                            // 更新上次发送邮件的时间
                            lastMailSendTime = currentTime;
                        }

                    }

                    List temp = redisUtil.getList("equipment:temperature:temperature" + serial);
                    List humidity = redisUtil.getList("equipment:temperature:humidity" + serial);
                    List gps_x = redisUtil.getList("equipment:temperature:gps_x" + serial);
                    List gps_y = redisUtil.getList("equipment:temperature:gps_y" + serial);
                    List time = redisUtil.getList("equipment:temperature:time" + serial);

                    SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
                    String timeString = sdf.format(new Date());
                    //将数据存入到redis
                    if (temp == null && humidity == null && time == null) {
//                    System.out.println("创建缓存");
                        List templist = new ArrayList();
                        templist.add(data.getInteger("temperature"));
                        redisUtil.setList("equipment:temperature:temperature" + serial, templist);
                        List humiditylist = new ArrayList();
                        humiditylist.add(data.getInteger("humidity"));
                        redisUtil.setList("equipment:temperature:humidity" + serial, humiditylist);
                        List timelist = new ArrayList();
                        timelist.add(timeString);
                        redisUtil.setList("equipment:temperature:time" + serial, timelist);
                        List gps_xlist = new ArrayList();
                        gps_xlist.add(data.getDouble("latitude"));
                        redisUtil.setList("equipment:temperature:gps_x" + serial, gps_xlist);
                        List gps_ylist = new ArrayList();
                        gps_ylist.add(data.getDouble("longitude"));
                        redisUtil.setList("equipment:temperature:gps_y" + serial, gps_ylist);
                    } else {
//                    System.out.println("已有缓存 读取缓存");
                        if (temp.size() < 10) {
//                        System.out.println("缓存小于10 添加");
                            temp.add(data.getInteger("temperature"));
                            redisUtil.setList("equipment:temperature:temperature" + serial, temp);
                            humidity.add(data.getInteger("humidity"));
                            redisUtil.setList("equipment:temperature:humidity" + serial, humidity);
                            time.add(timeString);
                            redisUtil.setList("equipment:temperature:time" + serial, time);
                            gps_x.add(data.getDouble("latitude"));
                            redisUtil.setList("equipment:temperature:gps_x" + serial, gps_x);
                            gps_y.add(data.getDouble("longitude"));
                            redisUtil.setList("equipment:temperature:gps_y" + serial, gps_y);
                        } else {
//                        System.out.println("缓存大于10 删除");
                            temp.remove(0);
                            temp.add(data.getInteger("temperature"));
                            redisUtil.setList("equipment:temperature:temperature" + serial, temp);
                            humidity.remove(0);
                            humidity.add(data.getInteger("humidity"));
                            redisUtil.setList("equipment:temperature:humidity" + serial, humidity);
                            time.remove(0);
                            time.add(timeString);
                            redisUtil.setList("equipment:temperature:time" + serial, time);
                            gps_x.remove(0);
                            gps_x.add(data.getDouble("latitude"));
                            redisUtil.setList("equipment:temperature:gps_x" + serial, gps_x);
                            gps_y.remove(0);
                            gps_y.add(data.getDouble("longitude"));
                            redisUtil.setList("equipment:temperature:gps_y" + serial, gps_y);
                        }
                        System.out.println("缓存数据" + serial + "：" + temp + " " + humidity + " " + time);
                    }
//                System.out.println("\nReceived Sensor Data:");
//                System.out.println("Temperature: " + data.getDouble("temperature") + "℃");
//                System.out.println("Humidity: " + data.getInteger("humidity") + "%");
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                }
            });

            // 循环监听终止条件
            while (running && !Thread.currentThread().isInterrupted()) {
                try {
                    Thread.sleep(1000); // 允许中断的休眠
                } catch (InterruptedException e) {
                    // 重新设置中断标志并退出循环
                    Thread.currentThread().interrupt();
                    running = false;
                    break;
                }
            }
        } finally {
            if (publisherClient != null && publisherClient.isConnected()) {
                try {
                    subscriberClient.disconnect();
                    subscriberClient.close();
                } catch (MqttException e) {
                    System.out.println(e.getMessage());
                }
            }
//            // 确保关闭订阅客户端
//            if (subscriberClient != null && subscriberClient.isConnected()) {
//                try {
////                    subscriberClient.unsubscribe(topic + serial);
////                    subscriberClient.disconnectForcibly(1000);
//                    subscriberClient.close();
//                } catch (MqttException e) {
//                    System.out.println("Error closing subscriber client: " + e.getMessage());
//                }
//            }
        }
    }
}
